package com.example.messagedemo.conifg;

import com.example.messagedemo.consumer.MessageConsumer;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ConsumerConfig {

    final private MessageConsumer messageConsumer;
    final private ClientConfig clientConfig;

    public ConsumerConfig(MessageConsumer messageConsumer, ClientConfig clientConfig) {
        this.messageConsumer = messageConsumer;
        this.clientConfig = clientConfig;
    }

    @Bean(destroyMethod = "shutdown")
    public DefaultMQPushConsumer consumer() {
        //指定订阅组名称
        String consumerGroup = clientConfig.getToken() + "_bcg_fast";
        // 使用指定的消费者组名称进行实例化。
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(null, consumerGroup , new AclClientRPCHook(new SessionCredentials(clientConfig.getAccessKey(), clientConfig.getSecretKey())));
        // 指定 NameServer 地址
        consumer.setNamesrvAddr(clientConfig.getNameSvrAddr());
        // consumer.setUseHeartbeatV2(true);
        // consumer.setMessageModel(MessageModel.BROADCASTING);
        // consumer.setVipChannelEnabled(true);
        consumer.setHeartbeatBrokerInterval(1000);
        System.out.println(consumer.getHeartbeatBrokerInterval());
        try {
            // 指定订阅的Token,subExpression:订阅过滤表达式字符串,默认订阅全部,示例1："tag1 || tag2 || tag3"
            consumer.subscribe(clientConfig.getToken(), "*");
            // 指定消息监听类
            consumer.registerMessageListener(messageConsumer);
            // 消息监听启动
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return consumer;
    }
}